package com.joker.demo.core.netty.client.handler;

import com.alibaba.fastjson2.JSON;
import com.joker.demo.core.domain.dto.MessageDTO;
import com.joker.demo.core.domain.dto.TaskCancelRequestDTO;
import com.joker.demo.core.domain.dto.TaskExecuteRequestDTO;
import com.joker.demo.core.domain.enums.JobRequestEnum;
import com.joker.demo.core.executor.JobExecutor;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 任务执行器客户端处理器
 *
 * @author feixiang.li
 * @since 2025/10/23
 */
@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    private static final Logger log = LoggerFactory.getLogger(NettyClientHandler.class);
    
    // 重连回调
    private Runnable reconnectCallback;
    private final AtomicBoolean reconnectTriggered = new AtomicBoolean(false);
    
    // 任务执行器
    private final JobExecutor jobExecutor;

    public NettyClientHandler(JobExecutor jobExecutor) {
        this.jobExecutor = jobExecutor;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("【执行器】-连接到【调度器】成功");
        // 重置重连标志
        reconnectTriggered.set(false);
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("【执行器】-与【调度器】连接断开");
        
        // 触发重连回调（只触发一次）
        if (reconnectCallback != null && reconnectTriggered.compareAndSet(false, true)) {
            log.info("触发重连机制");
            reconnectCallback.run();
        }
        
        super.channelInactive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("【执行器】step 1: 接收到消息:{}", msg);
        // 处理自定义协议消息
        try {
            log.info("【执行器】收到自定义协议消息：{}", msg);

            // 解析消息
            MessageDTO request = JSON.parseObject(msg.toString(), MessageDTO.class);
            JobRequestEnum byCode = JobRequestEnum.getByCode(request.getType());
            if (byCode == null) {
                log.warn("未知消息类型：{}", request.getType());
                return;
            }

            // 处理不同类型的消息
            switch (byCode) {
                case REGISTER_RESPONSE:
                    handleRegisterResponse(ctx, request);
                    break;
                case HEARTBEAT_RESPONSE:
                    handleHeartbeatResponse(ctx, request);
                    break;
                case EXECUTE_TASK:
                    handleExecuteTask(ctx, request);
                    break;
                case CANCEL_TASK:
                    handleCancelTask(ctx, request);
                    break;
                case THREAD_POOL_MONITOR_RESPONSE:
                    handleThreadPoolMonitorResponse(ctx, request);
                    break;
                case ENVIRONMENT_MONITOR_RESPONSE:
                    handleEnvironmentMonitorResponse(ctx, request);
                    break;
                default:
                    log.warn("未知消息类型：{}", request.getType());
            }

        } catch (Exception e) {
            log.error("【执行器】处理自定义协议消息异常", e);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("【执行器】任务执行器客户端异常", cause);
        
        // 触发重连回调（只触发一次）
        if (reconnectCallback != null && reconnectTriggered.compareAndSet(false, true)) {
            log.info("因异常触发重连机制");
            reconnectCallback.run();
        }
        
        ctx.close();
    }

    /**
     * 设置重连回调
     */
    public void setReconnectCallback(Runnable reconnectCallback) {
        this.reconnectCallback = reconnectCallback;
    }

    /**
     * 处理注册响应
     */
    private void handleRegisterResponse(ChannelHandlerContext ctx, MessageDTO request) {
        log.info("【执行器】收到注册响应：{}", request.getContent());
    }

    /**
     * 处理心跳响应
     */
    private void handleHeartbeatResponse(ChannelHandlerContext ctx, MessageDTO request) {
        log.debug("【执行器】收到心跳响应：{}", request.getContent());
    }

    /**
     * 处理执行任务请求
     */
    private void handleExecuteTask(ChannelHandlerContext ctx, MessageDTO request) {
        try {
            log.info("【执行器】处理执行任务请求：{}", request.getContent());

            // 解析任务执行请求
            TaskExecuteRequestDTO taskRequest = JSON.parseObject(request.getContent(), TaskExecuteRequestDTO.class);

            if (taskRequest == null || taskRequest.getJobLogId() == null) {
                log.error("执行任务请求参数无效");
                sendExecuteTaskErrorResponse(ctx, request, "执行任务请求参数无效");
                return;
            }

            log.info("开始执行任务，任务ID：{} 运行记录id:{}", taskRequest.getJobId(), taskRequest.getJobLogId());

            // 提交任务到执行器
            jobExecutor.submitTask(taskRequest);

            // 发送执行成功响应
            MessageDTO response = new MessageDTO(JobRequestEnum.EXECUTE_TASK_RESPONSE.getCode(), request.getMessageId(), "任务正在执行");
            ctx.writeAndFlush(JSON.toJSONString(response));
            
            log.info("任务执行请求处理完成，任务ID：{}", taskRequest.getJobId());
            
        } catch (Exception e) {
            log.error("处理执行任务请求异常", e);
            sendExecuteTaskErrorResponse(ctx, request, "处理执行任务请求异常: " + e.getMessage());
        }
    }

    /**
     * 处理取消任务请求
     */
    private void handleCancelTask(ChannelHandlerContext ctx, MessageDTO request) {
        try {
            log.info("【执行器】处理取消任务请求：{}", request.getContent());

            // 解析取消任务请求
            TaskCancelRequestDTO cancelRequest = JSON.parseObject(request.getContent(), TaskCancelRequestDTO.class);

            if (cancelRequest == null || cancelRequest.getJobLogId() == null) {
                log.error("取消任务请求参数无效");
                sendCancelTaskErrorResponse(ctx, request, "取消任务请求参数无效");
                return;
            }

            log.info("开始取消任务，任务ID：{}", cancelRequest.getJobLogId());

            // 取消任务
            boolean cancelResult = jobExecutor.cancelTask(cancelRequest.getJobLogId());
            log.info("任务取消结果:{}", cancelResult);

            // 发送取消成功响应
            MessageDTO response = new MessageDTO(JobRequestEnum.CANCEL_TASK_RESPONSE.getCode(), request.getMessageId(), "任务取消成功");
            ctx.writeAndFlush(JSON.toJSONString(response));
            
            log.info("任务取消请求处理完成，任务ID：{}", cancelRequest.getJobLogId());
            
        } catch (Exception e) {
            log.error("处理取消任务请求异常", e);
            sendCancelTaskErrorResponse(ctx, request, "处理取消任务请求异常: " + e.getMessage());
        }
    }

    /**
     * 发送执行任务错误响应
     */
    private void sendExecuteTaskErrorResponse(ChannelHandlerContext ctx, MessageDTO originalRequest, String errorMessage) {
        try {
            MessageDTO errorResponse = new MessageDTO(JobRequestEnum.EXECUTE_TASK_RESPONSE.getCode(), originalRequest.getMessageId(), "执行任务失败: " + errorMessage);
            ctx.writeAndFlush(JSON.toJSONString(errorResponse));
        } catch (Exception e) {
            log.error("发送执行任务错误响应失败", e);
        }
    }

    /**
     * 发送取消任务错误响应
     */
    private void sendCancelTaskErrorResponse(ChannelHandlerContext ctx, MessageDTO originalRequest, String errorMessage) {
        try {
            MessageDTO errorResponse = new MessageDTO(JobRequestEnum.CANCEL_TASK_RESPONSE.getCode(), originalRequest.getMessageId(), "取消任务失败: " + errorMessage);
            ctx.writeAndFlush(JSON.toJSONString(errorResponse));
        } catch (Exception e) {
            log.error("发送取消任务错误响应失败", e);
        }
    }

    /**
     * 处理线程池监控数据响应
     */
    private void handleThreadPoolMonitorResponse(ChannelHandlerContext ctx, MessageDTO request) {
        log.debug("【执行器】收到线程池监控数据响应：{}", request.getContent());
    }

    /**
     * 处理环境监控数据响应
     */
    private void handleEnvironmentMonitorResponse(ChannelHandlerContext ctx, MessageDTO request) {
        log.debug("【执行器】收到环境监控数据响应：{}", request.getContent());
    }
}